package com.atguigu.gmall.realtime.dws.app;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class RefundSkuType {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
        //1.2 设置并行度
        env.setParallelism(1);
        //1.3 指定表执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));

        //退单
        Table orderRefundInfo = tEnv.sqlQuery(
                "select " +
                        "data['id'] id," +
                        "data['user_id'] user_id," +
                        "data['order_id'] order_id," +
                        "data['sku_id'] sku_id," +
                        "data['refund_reason_type'] refund_reason_type," +
                        "data['refund_reason_txt'] refund_reason_txt," +
                        "data['create_time'] create_time," +
                        "pt," +
                        "ts " +
                        "from topic_db " +
                        "where `table`='order_refund_info' " +
                        "and `type`='insert' ");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        // 3. 过滤订单表中的退单数据: order_info  update
        Table orderInfo = tEnv.sqlQuery(
                "select " +
                        "data['id'] id," +
                        "data['province_id'] province_id," +
                        "`old` " +
                        "from topic_db " +
                        "where `table`='order_info' " +
                        "and `type`='update'" +
                        "and `old`['order_status'] is not null " +
                        "and `data`['order_status']='1005' ");
        tEnv.createTemporaryView("order_info", orderInfo);

        // 4. join: 普通的和 lookup join
        Table refund_total = tEnv.sqlQuery(
                "select " +
                        "ri.sku_id," +
                        "ri.refund_type," +
                        "ri.ts " +
                        "from order_refund_info ri " +
                        "join order_info oi " +
                        "on ri.order_id=oi.id ");
        tEnv.createTemporaryView("refund_total", refund_total);

        //计算10秒内各商品退款订单类型数
        tEnv.executeSql("SELECT window_start, window_end,sku_id,refund_type, count(*)  as refund_skutype_ct \n" +
                "  FROM TABLE(\n" +
                "    TUMBLE(TABLE refund_total, DESCRIPTOR(ts), INTERVAL '10' Seconds))\n" +
                "  GROUP BY window_start, window_end,sku_id,refund_type;").print();
    }
}
